package com.dtruth.dataflow.bolt;

import java.util.Map;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.thrift.TException;
import org.slf4j.Logger;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

import com.dtruth.dataflow.util.ETLServiceClient;
import com.dtruth.dataflow.util.Env;
import com.dtruth.dataflow.util.MyLoggerFactory;
import com.dtruth.dataflow.util.StringUtil;
import com.feheadline.cms.general.etl.service.thrift.gen.Article;
import com.feheadline.cms.general.etl.service.thrift.gen.ETLThriftService.Client;
import com.feheadline.cms.general.etl.service.thrift.gen.Message;

public class DataPushOutBolt extends BaseBasicBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	private String address = null;
	private static Logger log = MyLoggerFactory
			.getLogger(DataPushOutBolt.class);

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("news"));
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		address = Env.getInstance().getProperty("data.pushout.url");
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		if (address != null && !address.equals("")) {
			@SuppressWarnings("unchecked")
			Map<String, Object> map = (Map<String, Object>) input.getValue(0);
			Article article = geneArticle(map);
			if (article != null) {
				ETLServiceClient esc = new ETLServiceClient();
				Client client = null;
				try {
					client = esc.getClient(address);
					Message message = client.pushArticle(article, true);
					if (!message.isSuccess())
						log.warn("push to index failure:" + map.get("url"));
					else
						log.debug("emit:" + map.get("url"));
				} catch (TException e) {
					log.error(ExceptionUtils.getStackTrace(e));
				} finally {
					if (client != null)
						esc.closeClient();
				}
			} else
				log.warn("push to index wrong:" + map.get("key").toString());
		} else {
			log.warn("can't read thrift configuration:");
		}

	}

	Article geneArticle(Map<String, Object> map) {
		Article article = new Article();
		Object obj = map.get("key");
		article.setNews_key(obj.toString());
		obj = map.get("url");
		article.setUrl(obj.toString());
		obj = map.get("title");
		if (obj != null)
			article.setTitle(obj.toString());
		obj = map.get("summary");
		if (obj != null) {
			article.setSummary(obj.toString());
		}
		obj = map.get("body");
		if (obj != null){
			article.setContent(obj.toString());
		}	
		obj = map.get("category");
		if (obj != null)
			article.setCategory(obj.toString());
		obj = map.get("pubdate");
		if (obj != null) {
			if (StringUtil.isNumeric(obj.toString()))
				article.setPub_time(Long.valueOf(obj.toString()));
			else
				log.warn(map.get("key") + ":pubdata=" + obj);
		}
		obj = map.get("author");
		if (obj != null)
			article.setAuthor(obj.toString());
		obj = map.get("origin");
		if (obj != null)
			article.setOrigin(obj.toString());
		obj = map.get("domain");
		if (obj != null)
			article.setDomain(obj.toString());
		obj = map.get("channel");
		if (obj != null)
			article.setChannel(obj.toString());
		obj = map.get("thumbnail");
		if (obj != null)
			article.setImg_thum_url(obj.toString());
		obj = map.get("quality");
		if (obj != null)
			article.setScore(Integer.parseInt(obj.toString()));
		else
			article.setScore(1);
		return article;
	}
}